#pragma once
#include "../common/net.hpp"
#include "../common/message.hpp"
#include <unordered_set>
/*在当前的项⽬中，我们也实现⼀个简单的发布订阅功能，该功能是围绕多个客⼾端与⼀个服务端来展开的。
即，任意⼀个客⼾端在发布或订阅之前先创建⼀个主题，⽐如在新闻发布中我们创建⼀个⾳乐新闻主题，哪些客⼾端希望能够收到⾳乐新闻相关的消息，则就订阅这个主题，服务端会建⽴起该主题与客⼾端之间的联系。
当某个客⼾端向服务端发布消息，且发布消息的⽬标主题是⾳乐新闻主题，则服务端会找出订阅了该主题的客⼾端，将消息推送给这些客⼾端。*/
namespace bitrpc
{
    namespace server
    {
        class TopicManager
        {
        public:
            using ptr = std::shared_ptr<TopicManager>;
            TopicManager() {}
            // 注册给dispatcher模块的，用于处理主题请求的回调函数
            void onTopicRequest(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                // 主题的创建
                // 主题的删除
                // 主题的订阅
                // 取消订阅
                // 主题消息发布

                TopicOptype optype = msg->topicOptype();
                bool ret = true;
                switch (optype)
                {
                case TopicOptype::TOPIC_CREATE:
                    topicCreate(conn, msg);
                    break;
                case TopicOptype::TOPIC_REMOVE:
                    topicRemove(conn, msg);
                    break;
                case TopicOptype::TOPIC_SUBSCRIBE:
                    ret = topicSubscribe(conn, msg);
                    break;
                case TopicOptype::TOPIC_CANCEL:
                    topicCancel(conn, msg);
                    break;
                case TopicOptype::TOPIC_PUBLISH:
                    ret = topicPublish(conn, msg);
                    break;
                default: return errorResponse(conn,msg,RCODE::RCODE_INVALID_OPTYPE);
                }
                if(ret == false)
                {
                    errorResponse(conn,msg,RCODE::RCODE_NOT_FOUND_TOPIC);
                }
                else
                {
                    topicResponse(conn,msg);
                }
            }
            // 注册给MuduoServer模块的，用于处理断开连接的回调函数
            void onShutDown(const BaseConnection::ptr &conn)
            {
                //消息发布者断开连接，不需要处理，订阅者断开连接，需要删除其关联数据

                //1.判断断开连接的是不是订阅者，如果不是则直接返回
                std::vector<Topic::ptr> topics;
                Subscriber::ptr subscriber;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    if(_subscribers.count(conn) == 0)
                    {
                        //断开连接的不是订阅者
                        return;
                    }
                    subscriber = _subscribers[conn];
                //2.获取到订阅者退出后，会受到影响的主题对象
                for(auto &topic_name : subscriber->topics)
                {
                    if(_topics.count(topic_name) == 0)
                    {
                        continue;
                    }
                    topics.push_back(_topics[topic_name]);
                }
                //4.从订阅者映射信息中，删除订阅者
                _subscribers.erase(conn);
                }
                //3.从这些主题对象中，移除订阅者
                for(auto &topic:topics)
                {
                    topic->removeScriber(subscriber);
                }
            }

        private:
            void errorResponse(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg,RCODE rode)
            {
                auto rsp = MessageFactory::create<TopicResponse>();
                rsp->SetId(msg->rid());
                rsp->SetMessageType(MessageType::RSP_TOPIC);
                rsp->setRcode(rode);
                conn->send(rsp);
            }
            // 对于服务注册的响应

            void topicResponse(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                auto rsp = MessageFactory::create<TopicResponse>();
                rsp->SetId(msg->rid());
                rsp->SetMessageType(MessageType::RSP_TOPIC);
                rsp->setRcode(RCODE::RCODE_OK);
                conn->send(rsp);
            }
            void topicCreate(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                // 这里，每次都创建主题，后一次的主题覆盖前一次的主题
                std::unique_lock<std::mutex> lock(_mutex);
                // 构造一个主题对象，添加映射关系的管理
                std::string topic_name = msg->TopicKey();
                auto topic = std::make_shared<Topic>(topic_name);
                // 换成insert，insert的含义就是，如果存在，则插入失败
                _topics.insert(std::make_pair(topic_name, topic));
                //_topics[topic_name] = topic;  // 因为你采用的[] 这个方式，这个方式会覆盖，也就是说topic_name存在的时候
                // topic替换之前的topic
                DLOG("主题创建成功");

            }
            void topicRemove(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                // 查看当前主题有哪些订阅者，然后从订阅者中删除主题信息
                // 删除主题的数据---主题名称与主题对象的映射关系
                std::string topic_name = msg->TopicKey(); // 先获取主题名称
                Topic::ptr topic;
                std::unordered_set<Subscriber::ptr> subscribers;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    if (_topics.count(topic_name) == 0)
                    {
                        // 没有该主题对象
                        return;
                    }
                    topic = _topics[topic_name];
                    subscribers = topic->_subscribers;
                    _topics.erase(topic_name); // 删除主题对象
                }
                for (auto &sub : subscribers)
                {
                    // 遍历该主题的所有订阅者，删除每个订阅者的该主题记录
                    sub->removeTopic(topic_name);
                }
                DLOG("主题删除成功");

            }
            bool topicSubscribe(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                // 1.先找出主题对象，以及订阅者对象（同一个链接可能有多个主题订阅）
                // 如果没有找到主题对象就报错，如果没找到订阅者对象，那就新增一个订阅者
                Topic::ptr topic;
                Subscriber::ptr subscriber;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    if (_topics.count(msg->TopicKey()) == 0)
                    {
                        DLOG("主题订阅失败");

                        return false;
                    }
                    topic = _topics[msg->TopicKey()]; // 找到了主题对象
                    if (_subscribers.count(conn))
                    {
                        // 找到了该链接的订阅者对象
                        subscriber = _subscribers[conn];
                    }
                    else
                    {
                        // 该连接没有订阅者对象，首次订阅
                        subscriber = std::make_shared<Subscriber>(conn);
                        _subscribers[conn] = subscriber;
                    }
                }
                // 2.在主题对象中，新增一个订阅者关联的链接；在订阅者对象中新增一个订阅的主题
                topic->appendScriber(subscriber);
                subscriber->appendTopic(msg->TopicKey());
                DLOG("主题订阅成功");

                return true;

            }
            void topicCancel(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                // 1.先找出主题对象，以及订阅者对象
                // 主题不存在或者订阅者不存在，则不处理
                Topic::ptr topic;
                Subscriber::ptr subscriber;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    if (_topics.count(msg->TopicKey()))
                    {
                        topic = _topics[msg->TopicKey()]; // 找到了主题对象
                    }
                    if (_subscribers.count(conn))
                    {
                        // 找到了该链接的订阅者对象
                        subscriber = _subscribers[conn];
                    }
                }
                // 2.从主题对象中删除当前的订阅者连接，从订阅者中删除订阅的主题
                if (topic && subscriber)
                    topic->removeScriber(subscriber);
                if (subscriber)
                    subscriber->removeTopic(msg->TopicKey());
            }
            bool topicPublish(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                // 先找到要发布的主题对象，通过该主题对象的所有订阅者连接向订阅者发送消息
                Topic::ptr topic;
                std::string name = msg->TopicKey();
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    if (_topics.count(name) == 0)
                    {
                        // 没找到主题对象
                        return false;
                    }
                    topic = _topics[name];
                }
                topic->pushMessage(msg);
                DLOG("主题消息发布成功,发送内容：%s",msg->topicMsg().c_str());

                return true;
            }

        private:
            struct Subscriber // 订阅者对象
            {
                using ptr = std::shared_ptr<Subscriber>;
                std::mutex _mutex;
                BaseConnection::ptr _conn;
                std::unordered_set<std::string> topics; // 该订阅者订阅了哪些主题

                Subscriber(const BaseConnection::ptr &conn)
                    : _conn(conn)
                {
                }
                // 订阅主题的时候调用
                void appendTopic(const std::string &topic_name)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    topics.insert(topic_name);
                }
                // 删除主题,当主题被删除或者取消订阅的时候被调用
                void removeTopic(const std::string &topic_name)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    topics.erase(topic_name);
                }
            };
            struct Topic // 主题对象
            {
                using ptr = std::shared_ptr<Topic>;
                std::mutex _mutex;
                std::string _topic_name;                          // 主题名称
                std::unordered_set<Subscriber::ptr> _subscribers; // 当前主题的所有订阅者
                Topic(const std::string &name)
                    : _topic_name(name)
                {
                }

                // 添加该主题的订阅者，新增订阅时调用
                void appendScriber(const Subscriber::ptr &subscriber)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    _subscribers.insert(subscriber);
                }
                // 删除该主题的订阅者，取消订阅或者订阅者断开连接时调用
                void removeScriber(const Subscriber::ptr &subscriber)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    _subscribers.erase(subscriber);
                }
                // 从该主题 发布消息，收到消息发布请求时调用
                void pushMessage(const BaseMessage::ptr &msg)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    for (auto &subscriber : _subscribers)
                    {
                        subscriber->_conn->send(msg);
                    }
                }
            };

        private:
            std::mutex _mutex;
            std::unordered_map<std::string, Topic::ptr> _topics;                   // 主题名称与主题对象的映射关系，主题对象中包含了该主题的所有订阅者
            std::unordered_map<BaseConnection::ptr, Subscriber::ptr> _subscribers; // 连接与订阅者的映射关系
        };

    }
}